//
// substation.cpp
// substation
// Application Server Substation / Substation acts as controller/passthru on a single machine
// Copyright (C) 2003, 2004, 2005 eXegeSys, Inc.
// Copyright (C) 2008 Bruce A. James
//
// This program is part of the Protea Project.
//
// The Protea Project is free software; you can redistribute it and/or modify it 
// under the terms of the GNU General Public License as published by the 
// Free Software Foundation, either version 3 of the License, or (at your 
// option) any later version.
//
// The Protea Project is distributed in the hope that it will be useful, but 
// WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY 
// or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License 
// for more details.
// 
// You should have received a copy of the GNU General Public License along 
// with this program.  If not, see <http://www.gnu.org/licenses/>.
// 
// Please send correspondence to:
// 
// theproteaproject@gmail.com
//

//
// substation.cpp
//
//

#include <signal.h>
#include "../defines.h"
#include "../entry.h"
#include "../AppHandler/AppInterface.h"
#include "../AppHandler/parms_xml.h"
#include "../OSDep/misc.h"
#include "../OSDep/service.h"
#include "substation.h"


DAEMON_SERVICE( XCSubstation, sub, "Substation", "Protea Substation", "Provides coordination for Protea Project applications running on this machine", "GCS\0\0" );



void catchchildren( int signum )
{
	XCWaitChildren( &sub, signum );
};


void abort( int signum ) {
	sub.ApplStop();
};



///////
//
// Substation class
//
XCSubstation::XCSubstation( char* name, char* dispname, char* desc, void* func, char* deps ) : XCService( name, dispname, desc, func, deps )
{
	murderer = NULL;
	want_interfaces = false;
};

XCSubstation::~XCSubstation()
{
};


void XCSubstation::SendGoodbye( int appnum, int pid ) {
	// Grab system hostname
	char* hostname = new char[MAX_PATH];
	XCGetHostName( hostname, MAX_PATH );

	// Create overall message
	ToolsTrace( TRC_APPINTERFACE, TLV_HIGH, "Create message" );
	XCToolsMessage hello( this );
	hello.CreateMsg( DISP_MESSAGE_SIZE + STARTUP_MESSAGE_SIZE + strlen( hostname ) + 1);
	hello.SetHeader( msg_dispatch, src_app, dest_gcs, false, true );

	// Formulate dispatch message
	ToolsTrace( TRC_APPINTERFACE, TLV_HIGH, "Move data into message" );
	disp_message* disp_info = (disp_message*)hello.GetDataPtr();
	disp_info->opcode = disp_goodbye;

	// Formulate startup message
	startup_message* startup = (startup_message*)disp_info->data;
	startup->appnum = myappnum;
	startup->pid = pid;
	startup->filelen = 0;
	sprintf( startup->data, "%s", process );
	delete [] hostname;

	// Send the result message
	SendMessage( msg_src, parent, &hello );
};



void XCSubstation::OnDroppedProcess( pid_t process_pid ) {
	process_info* pi = XCGetProcess( process_list, process_pid );
	if (pi) {
		printf( "A child died:\n" );
		printf( "Name: %s\n", pi->file );
		printf( "Number: %d\n", pi->startup.appl_nbr );

		bool is_active = pi->active;
		app_start_message tempappl = pi->startup;

		// Drop the previous process
		XCAppInterface::OnDroppedProcess( process_pid );

		// Send drop message
		SendGoodbye( tempappl.appl_nbr, process_pid );

		if (is_active) {
			printf( "Attempt to restart\n" );

			// Start the new process
			StartChildApp( &tempappl );
		};
	} else {
		// Forward on to default handler
		XCAppInterface::OnDroppedProcess( process_pid );
	};
};



void XCSubstation::OnDroppedConnection( XCMsgPortPtr tempport ) {
	printf( "Connection dropped!\n" );

	// If this port is associated with a process_info record that is inactive, delete it
	process_info* process = tempport->myprocess;
	if (process) {
		process->port = NULL;
		if (!process->active)
			XCAppInterface::OnDroppedProcess( process->pid );
	};

	// If the port was gcs, then stop execution
	if (parent == tempport) {
		printf( "GCS dropped connection, abort\n" );
		ApplStop();
	};
};


bool XCSubstation::SubstationForwardMessage( XCMsgPort* port, XCToolsMessage* msg ) {
	INFUNC( XCSubstation::SubstationForwardMessage, this );

	base_message* base = msg->GetHeaderPtr();
	bool query_message = (base->miscflag & misc_query) > 0;

	if (base->sdflag & dest_app) {
		trans_message* trans = (trans_message*)msg->GetDataPtr();
		long appnbr = trans->appnbr;

		// Determine correct child...
		XCMsgPortPtr dest = childlist;
		while ( dest && (long)dest->extradata != appnbr ) {
			dest = dest->next;
		};

		if (dest)
			ForwardMessage( port, msg, dest );
		else {
			ToolsTrace( TRC_APPINTERFACE, TLV_LOW, "No application to accept message" );
			RAISETOOLSERROR( 1 );
		};
	} else {
		ToolsTrace( TRC_TRAFFIC, TLV_MEDIUM, "Forward transaction towards gcs: %d\n", query_message );
		ForwardMessage( port, msg );
	};

	OUTFUNCRET( true );
};


//
// Callback Functions
//
/////////////////////////////////////////////////////////////////////////////////////
bool XCSubstation::GotMessage( XCMsgPort* port, XCToolsMessage* msg )
{
	bool retval = true;
	INFUNC( XCSubstation::GotMessage, this )
	
	DebugMessageTraffic( "Got message from ", port );

	disp_message* dmsg;
	XCMsgPortPtr tempport;
	XCMsgPortPtr outport = NULL;
	bool all_gone;

	msg_recv++;

	XCToolsMessage msg_send( this );

	base_message* base = msg->GetHeaderPtr();

	ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Substation Received Message : %d", base->type );
	msg->Dump();

	// Do we need to broadcast to all children?
	if (base->miscflag & misc_broadcast)
		BroadcastMessage( port, msg );
/*
	// Forward all messages intended for GCS
	if (base->sdflag & dest_gcs) {
		SendMessage( port, parent, msg );
		return true;
	};
*/
	switch( base->type ) {
		case msg_status:
		case msg_admin:
			// Forward status/admin messages to gcs
			ForwardMessage( port, msg );
			break;

		case msg_trans:
		case msg_sql_query:
			// Forward trans to gcs
//			printf( "Substation message forward!\n" );
			SubstationForwardMessage( port, msg );
			break;

		case msg_options:
			XCAppInterface::GotMessage( port, msg );
			break;

		case msg_dispatch:
			contact_ui_message* temp_msg;
			dmsg = (disp_message*)msg->GetDataPtr();
			app_start_message* tempappl;
			ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Dispatch Message Opcode : %ld" , dmsg->opcode );

			switch( dmsg->opcode )
			{
			case disp_pushdown_query:
				// Request for status
				SendStatus( src_app, parent );
				break;

			case disp_start_app:
				tempappl = (app_start_message*)dmsg->data;
				StartChildApp( tempappl );
				break;

			case disp_hello:
				if (base->sdflag & src_app)
				{
					// (not implemented) ensure all dispatchers are started before sending the message along...
					// mark as ready dispatcher and forward the message to gcs
					if (port->myprocess)
						port->myprocess->active = true;
					port->ready = true;
					ForwardMessage( port, msg );
				}
				break;

			case disp_contact_ui:
				temp_msg = (contact_ui_message*)dmsg->data;
				ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Substation receives UI contact request %ld %s %ld %s", temp_msg->appnum, temp_msg->appname, temp_msg->port, temp_msg->host );
				tempport = childlist;
				while (tempport != NULL)
				{
					if ( tempport->porttype == pt_disp && tempport->extradata == (void*)temp_msg->appnum )
						outport = tempport;
					ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Checking Child : %ld, type : %d, AppNum : %d == %d?", tempport->transport->GetOutHandle(), tempport->porttype, tempport->extradata, temp_msg->appnum );
					tempport = tempport->next;
				};
				if (!outport) {
					printf( "Unable to find application: %ld\n", temp_msg->appnum );
					RAISETOOLSERROR( -1 );
				};
				SendMessage( port, outport, msg );
				break;

			case disp_shutdown:
				// if a broadcast, no need to send the message along
				// TODO: Implement
				if (!(base->miscflag & misc_broadcast)) {
					tempport = childlist;
				};

				murderer = parent;
				break;

			case disp_shutdown_confirm:

				if (base->miscflag & misc_broadcast)
				{
					all_gone = true;	// check if all our children are gone!!!


					// Loop thru the list of children
					outport = NULL;
					tempport = childlist;
					while (tempport) {
						if ( tempport == port ) {
							// This port is the port the message arrived from
							tempport->ready = false;
							tempport->transport->Close();
							// delete XCMsgPortPtr
							if (outport) {
								outport->next = tempport->next;
								delete tempport;
								tempport = outport->next;
							} else {
								childlist = tempport->next;
								delete tempport;
								tempport = childlist;
							};

							// We have deleted the port that sent us the message, so return false
							retval = false;
						} else if ( tempport == parent ) {
							// Found our parent port
							outport = tempport;
							tempport = tempport->next;
						} else if ( tempport->ready ) {
							all_gone = false;
							outport = tempport;
							tempport = tempport->next;
						}
						else
						{
							outport = tempport;
							tempport = tempport->next;
						};
					};
					// all my children are dead!!! (sob!) okay, then I might as well 
					// commit suicide... of course I'll tell the murderer about it first!
					if (all_gone)
					{
						if (murderer) {
//							ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "All children are gone - murderer : %ld", murderer->transport->GetInHandle() );

							msg_send.CreateMsg( DISP_MESSAGE_SIZE );
							msg_send.SetHeader( msg_dispatch, src_app, dest_gcs, true, false);
							dmsg = (disp_message*)msg_send.GetDataPtr();
							dmsg->opcode = disp_shutdown_confirm;
							SendMessage( msg_src, murderer, &msg_send );
							msg_send.DeleteMsg();
							murderer->transport->Close();
						};
						ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Substation Stopping Now" );
						QuitLoop();
					};
				};
				break;

			case disp_abort_txn:
				dmsg = (disp_message*)msg->GetDataPtr();
				tempport = childlist;
				while (tempport && tempport->extradata != (void*)dmsg->appnbr )
					tempport = tempport->next;

				if (!tempport) {
					printf( "Unable to find application: %ld\n", dmsg->appnbr );
					RAISETOOLSERROR( -1 );
				};
				SendMessage( port, tempport, msg );
				break;

			default:
				ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Invalid Dispatch Message" );
				break;
			}
			break;

		default:
			ToolsTrace( TRC_TRAFFIC, TLV_HIGH, "Incorrect message type" );
			break;
	};

	OUTFUNCRET( retval )
};






void XCSubstation::StartMessage()
{
	const char* subname = "substation";

	// Send message to substation ... gcs, denoting readiness
	parent->porttype = pt_disp;

	// Create overall message
	ToolsTrace( TRC_APPINTERFACE, TLV_HIGH, "Create message" );
	XCToolsMessage hello( this );
	hello.CreateMsg( DISP_MESSAGE_SIZE + STARTUP_MESSAGE_SIZE + strlen( hostname ) + strlen( subname ) + 1);
	hello.SetHeader( msg_dispatch, src_app, dest_gcs, false, true );

	// Formulate dispatch message
	ToolsTrace( TRC_APPINTERFACE, TLV_HIGH, "Move data into message" );
	disp_message* disp_info = (disp_message*)hello.GetDataPtr();
	disp_info->opcode = disp_hello;

	// Formulate startup message
	startup_message* startup = (startup_message*)disp_info->data;
	startup->appnum = -1;
	startup->pid = XCGetProcessIdentifier();
	startup->filelen = strlen( subname );
	sprintf( startup->data, "%s%s", subname, hostname );

	// Send the result message
	ToolsTrace( TRC_APPINTERFACE, TLV_HIGH, "Send message" );
	SendMessage( msg_src, parent, &hello );
};



void XCSubstation::ApplStop()
{
    INFUNC( XCSubstation::ApplStop, this )
	int count = 0;

    XCToolsMessage msg_send( this );
    disp_message* disp_info;

	msg_send.CreateMsg( DISP_MESSAGE_SIZE );
	msg_send.SetHeader( msg_dispatch, src_gcs, dest_app, true, false );
	disp_info = (disp_message*)msg_send.GetDataPtr();
	disp_info->opcode = disp_shutdown;

	sub.childlistlock.Lock();
	XCMsgPortPtr tempport = childlist;
	while (tempport) {
		try {
			if (tempport != parent) {
				count++;
				SendMessage( msg_src, tempport, &msg_send );
			};
		} catch( ... ) {};

		tempport = tempport->next;
	};
	sub.childlistlock.UnLock();

	// If we have no child, then we can just abort
	if (!count)
		sub.QuitLoop();

    OUTFUNC()
};




void XCSubstation::StartSubstation() {
	GetParent()->porttype = pt_gcs;
	GetParent()->ready = true;

	// Query GCS for load balancing parameters
	sub.hostname = new char[MAX_PATH];
	XCGetHostName( sub.hostname, MAX_PATH );

	// Send the start message for the substation
	StartMessage();

	XCToolsMessage hello( this );
	hello.CreateMsg( DISP_MESSAGE_SIZE + strlen( sub.hostname ) + 1);
	hello.SetHeader( msg_dispatch, src_gcs, dest_gcs, false, true );
	disp_message* disp_info = (disp_message*)hello.GetDataPtr();
	disp_info->opcode = disp_query_info;
	strcpy( disp_info->data, sub.hostname );
	SendMessage( msg_src, GetParent(), &hello );

	// Send Status Message to GCS
	process = strdup( "substation" );
	parentpid = 0;
	SendStatus( src_gcs, GetParent() );

	// Start child applications
	StartApps();
	delete [] hostname;
};





void XCSubstation::StartChildApp( app_start_message* tempappl ) {
	printf( "Substation start child, debug = %d, app=%s\n", tempappl->debug_type_filter, tempappl->appl_name );

	XCMsgPortPtr outport;
	if (! tempappl->NETapp )
	{
		outport = CreateChild( "driver", 0, NULL, true, tempappl );
	}
	else
	{
		outport = CreateChild( "NETDriver", 0, NULL, true, tempappl );
	}

	outport->porttype = pt_disp;
	outport->ready = false;
	outport->extradata = (void*)tempappl->appl_nbr;

	XCToolsMessage* msg = new XCToolsMessage( this );
	msg->CreateMsg( DISP_MESSAGE_SIZE + sizeof( app_start_message ) );
	msg->SetHeader( msg_dispatch, src_gcs, dest_app, false, false );

	disp_message* disp_info = (disp_message*)msg->GetDataPtr();
	disp_info->opcode = disp_start_app;
	app_start_message* start_message = (app_start_message*)disp_info->data;

	memcpy( start_message, tempappl, sizeof( app_start_message ) );

	// send start application message to dispatcher
	SendMessage( msg_src, outport, msg );

//	msg->Release();
};



void XCSubstation::StartApps() {
	app_start_message message;

	XCStartApp* app = applist;
	int count = 0;

	while (app) {
		printf( "Attempt app start: %s on %s\n", app->name, app->machine );
		if (XCStrCaseCompare( app->machine, hostname )==0) {
			memset( &message, 0, sizeof( app_start_message ) );
			message.appl_nbr = app->number;
			strcpy( message.appl_name, app->name );
			memcpy( message.debug_level_cutoff, app->debug_level_cutoff, sizeof( int ) * num_tc );
			message.debug_type_filter = app->debug_type_filter;
			message.authentication_method = auth_method;
			message.worker_threads = worker_threads;
			message.NETapp = app->NETapp;

			int offset = 0;
			if (auth_parameter || gcs.auth_parameter) {
				// Use gcs host value if none supplied for the application
				char* parm = auth_parameter ? auth_parameter : gcs.auth_parameter;
				message.authentication_parm_len = strlen( parm );
				memcpy( message.data, parm, message.authentication_parm_len );
				offset += message.authentication_parm_len;
			};

			if (app->smtp_host || gcs.smtp_host) {
				// Use gcs host value if none supplied for the application
				char* host = app->smtp_host ? app->smtp_host : gcs.smtp_host;
				message.smtp_len = strlen( host );
				memcpy( message.data+offset, host, message.smtp_len );
				offset += message.smtp_len;
			};

			if (app->printer || gcs.printer) {
				// Use gcs printer value if none supplied
				char* printer = app->printer ? app->printer : gcs.printer;
				message.printer_len = strlen( printer );
				memcpy( message.data+offset, printer, message.printer_len );
				offset += message.printer_len;
			};

			if (app->admin_email || gcs.admin_email) {
				// Use gcs admin value if none supplied
				char* admin = app->admin_email ? app->admin_email : gcs.admin_email;
				message.admin_email_len = strlen( admin );
				memcpy( message.data+offset, admin, message.admin_email_len );
				offset += message.admin_email_len;
			};

			if (app->fop_location || gcs.fop_location) {
				// Use gcs loc value if none supplied
				char* loc = app->fop_location ? app->fop_location : gcs.fop_location;
				message.fop_location_len = strlen( loc );
				memcpy( message.data+offset, loc, message.fop_location_len );
				offset += message.fop_location_len;
			};

			printf( "Start app: %s\n", app->name );
			StartChildApp( &message );
			count++;
		};

		app = app->next;
	};

	if (!count)
		printf( "WARNING - SUBSTATION STARTED NO APPLICATIONS!\n" );
};





//
// Service Functions
//
/////////////////////////////////////////////////////////////////////////////////////
void XCSubstation::OnInit() {
	// Check security
	DO_TRAFFIC_SECURITY_CHECK;

	// Turn on debugging if necessary
	XCAppOptions myopts;
	sub.GetOptions( myopts );
	myopts.debug_type_filter = TRC_APPINTERFACE | TRC_TRAFFIC | TRC_MESSAGING | TRC_FUNCTIONS;
	myopts.debug_type_filter = 0;
	myopts.debug_level_cutoff[ TCL_APPINTERFACE ] = TLV_EVERYTHING;
	myopts.debug_level_cutoff[ TCL_TRAFFIC ] = TLV_EVERYTHING;
	myopts.debug_level_cutoff[ TCL_MESSAGING ] = TLV_EVERYTHING;
	myopts.debug_level_cutoff[ TCL_FUNCTIONS ] = TLV_EVERYTHING;
	sub.SetOptions( myopts );
};

void XCSubstation::OnRun() {
	printf( "Start sub\n" );
	sub.StartSubstation();
	printf( "Message loop\n" );
	sub.MessageLoop();
};

void XCSubstation::OnSignalStop() {
	sub.ApplStop();
};

void XCSubstation::OnCleanup() {
	cout << "Waiting for children to exit..." << endl;
	XCWaitForChildrenToExit();
	cout << "All children finished..." << endl;
};



void XCSubstation::SetParms() {
	auth_method = (AuthenticateMethod)gcs.auth_method;
	worker_threads = gcs.worker_threads;
};



//
// substation Main function
//
/////////////////////////////////////////////////////////////////////////////////////
int main( int argc, char** argv )
{
	Protea::ShowIntro( "substation" );

	if (argc > 1 && !strcmp(argv[1],"-L"))
	{
		Protea::ShowLicense();
		return 0;
	};

	Protea::Initialize();

	try
	{
		INFUNC( main, NULL )
		DO_TRAFFIC_SECURITY_CHECK;

		XCServiceDispatcher daemon;

		if ( argc != 1 && argc != 2 && argc != 4 )
		{
			cout << "Program may only be executed through GCS" << endl;
			return 1;
		}

		// Usage : substation <gcs_host_ip> <gcs_sub_listen_port> <pipe>
		// currently <pipe> is ignored...

		// Register our signal handlers
		OSDEP_SIGNAL( SIGCHLD, catchchildren );
		OSDEP_SIGNAL( SIGTERM, abort );
	
		sub.StartLogging( argv[0] );
		if (!sub.LoadParms( argv[0] )) {
			cout << "Unable to load gcs.xml file!" << endl;
		};

		// Create socket connection
		if (argc == 2) {
			if (strcmp( argv[1], "install" ) ==0) {
				sub.Install();
				RETURNFUNC( 0 );
			} else if (strcmp( argv[1], "uninstall" )==0) {
				sub.Uninstall();
				RETURNFUNC( 0 );
			};
			RETURNFUNC( 1 );
		} else if (argc == 4) {
			init_message init;
			sub.StartApp( (PORT_HANDLE)atoi( argv[3] ), dest_gcs, init );
			daemon.SetService( false );
		} else {
			daemon.SetService( true );
			sub.SetParent( sub.Connect( sub.gcs.machine, sub.gcs.port ) );

			// Set startup parameters
			sub.SetParms();
		};

		// Run the service
		daemon.Add( &sub );
		daemon.Register();

		OUTFUNC()
	} catch( XCToolsError& error ) {
		XCOutputCallStack( &sub, error.GetErrorNumber(), error.GetDescription() );
	} catch( ... ) {
		XCOutputCallStack( &sub, -1, NULL );
	};

	Protea::Uninitialize();

	return 0;
};
